package com.atgugu.flink.chapter07.timer;

import com.atgugu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/4/6 10:02
 */
public class Flink03_Timer_Practice_1 {
    public static void main(String[] args) throws Exception {
        
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 当水位超过20的时候, 5s之后发出预警
        env
            .socketTextStream("hadoop162", 9999)
            .map(line -> {
                String[] data = line.split(",");
                return new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2]));
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((ws, ts) -> ws.getTs())
            
            )
            .keyBy(WaterSensor::getId) // 如果要用定时器, 必须先keyBy 因为定时器是和key相关
            .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                // 成员变量的个数和并行度相等
                /*private long ts;  // 保存定时器的时间的
                private int lastVc;  // 上一次的水位
                boolean isFirst = true;*/
                
                Map<String, Long> tsMap = new HashMap<>();
                Map<String, Integer> lastVcMap = new HashMap<>();
                Map<String, Boolean> isFirstMap = new HashMap<>();
                
                // 当定时器触发的执行这个方法
                @Override
                public void onTimer(long timestamp,  // 定时器触发的时间
                                    OnTimerContext ctx,
                                    Collector<String> out) throws Exception {
                    out.collect(ctx.getCurrentKey() + " 水位练习5s上升,发出预警信息....");
//                    isFirstMap.put(ctx.getCurrentKey(), null);
                    isFirstMap.remove(ctx.getCurrentKey());
                }
                
                @Override
                public void processElement(WaterSensor value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    // 5s内水位连续上升, 则预警
                    /*
                    当第一条数据进来的时候, 定义一个5s后触发的定时器
                    如果下一条数据是上升, 则什么都不做
                    如果下一条数据不是上升, 则删除定义定时器, 然后重新注册一个新的定时器
                     */
                    if (isFirstMap.get(ctx.getCurrentKey()) == null) { // 是第一条
                        // 注册定时器
                        // 提升成员变量: shift+ctrl+f
                        long ts = value.getTs() + 5000;
                        ctx.timerService().registerEventTimeTimer(ts);
                        tsMap.put(ctx.getCurrentKey(), ts);
                        System.out.println(ctx.getCurrentKey() + " 第一条数据过来, 注册定时器: " + ts);
                        // 把标记设为false
                        isFirstMap.put(ctx.getCurrentKey(), false);
                    } else {   //不是第一条
                        // 如果下降, 则删除定时器, 如果是上升或者不变, 则什么都不做
                        if (value.getVc() < lastVcMap.get(ctx.getCurrentKey())) {
                            
                            System.out.println(ctx.getCurrentKey() + " 水位下降, 删除定时器: " + tsMap.get(ctx.getCurrentKey()));
                            ctx.timerService().deleteEventTimeTimer(tsMap.get(ctx.getCurrentKey()));
                            
                            // 重新注册定时器
                            long ts = value.getTs() + 5000;
                            ctx.timerService().registerEventTimeTimer(ts);
                            tsMap.put(ctx.getCurrentKey(), ts);
                            System.out.println(ctx.getCurrentKey() + " 重新注册新的定时器: " + ts);
                        } else {
                            System.out.println(ctx.getCurrentKey() + " 水位没有下降, 则什么都不做: ");
                        }
                    }
                    
                    lastVcMap.put(ctx.getCurrentKey(), value.getVc());
                    
                }
            })
            .print();
        
        
        env.execute("flink_timer");
        
        
    }
}
